package cn.jly.flink.source2sink.redis;

import cn.jly.flink.entity.Person;
import cn.jly.flink.sideoutput.MyRedisCommandContainer;
import cn.jly.flink.sideoutput.impl.MyRedisCommandContainerImpl;
import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import redis.clients.jedis.HostAndPort;

import java.util.HashSet;
import java.util.Map;

/**
 * @PackageName cn.jly.flink.source2sink.redis
 * @ClassName FlinkRedisSourceDemo
 * @Description redis source
 * @Author 姬岚洋
 * @Date 2021/1/15 下午 4:39
 */
public class FlinkRedisSourceDemo {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.addSource(new RedisSource(Person.class.getSimpleName()))
                .print();

        env.execute("FlinkRedisSourceDemo");
    }

    public static class RedisSource extends RichSourceFunction<Person> {
        // 不使用原生的jedisCluster操作，因为再flink算子中传输，最好实现序列化
        // JedisCluster jedisCluster = null;
        MyRedisCommandContainer myRedisCommandContainer = null;

        /**
         * 读取redis的哪个key
         */
        private final String redisHashKey;

        public RedisSource(String redisHashKey) {
            this.redisHashKey = redisHashKey;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);

            final HashSet<HostAndPort> set = new HashSet<>(6);
            String ip = "192.168.172.101";
            set.add(new HostAndPort(ip, 7001));
            set.add(new HostAndPort(ip, 7002));
            set.add(new HostAndPort(ip, 7003));
            set.add(new HostAndPort(ip, 7004));
            set.add(new HostAndPort(ip, 7005));
            set.add(new HostAndPort(ip, 7006));
            myRedisCommandContainer = new MyRedisCommandContainerImpl(set);
        }

        @Override
        public void run(SourceContext<Person> ctx) throws Exception {
            final Map<String, String> map = myRedisCommandContainer.hgetAll(redisHashKey);
            for (Map.Entry<String, String> entry : map.entrySet()) {
                if (StringUtils.isNotEmpty(entry.getValue())) {
                    ctx.collect(JSON.parseObject(entry.getValue(), Person.class));
                }
            }
        }

        @Override
        public void cancel() {
            if (myRedisCommandContainer != null) {
                myRedisCommandContainer.close();
            }
        }
    }
}
